﻿#include "threadPoolReview.h"

/* threadPool类 */
// 构造函数
ThreadPool::ThreadPool()
	:initThreadPoolSize_(INIT_POOL_SIZE),
	threadPoolMaxSize_(THREAD_MAX_SIZE),
	curThreadPoolSize_(0),
	idleThreadPoolSize_(0),
	taskQueMaxSize_(TASK_QUE_MAX_SIZE),
	curTaskQueSize_(0),
	mode_(Pool_Mode::FIXED_MODE),
	isPoolRunning_(false)
{

}
// 析构函数
ThreadPool::~ThreadPool()
{
	// 开始回收资源
	// 将线程池关闭
	this->isPoolRunning_ = false;
	// 获取任务队列的互斥锁
	std::unique_lock<std::mutex> lock(taskQueMtx_);
	// 通知所有等待在notempty变量的线程
	notEmpty_.notify_all();
	// 等待线程池容器为空
	//isExit_.wait(lock, [&]() { return !threadList_.size(); });
	isExit_.wait(lock, [&]() { return !curThreadPoolSize_; });
}
// 设置线程池模式
void ThreadPool::setPoolMode(Pool_Mode mode)
{
	this->mode_ = mode;
}
// cached模式下设置最大线程数量
void ThreadPool::setThreadPoolMaxSize(int size)
{
	if (this->checkPoolRunning() || 
		this->mode_ == Pool_Mode::FIXED_MODE)
		return;
	this->threadPoolMaxSize_ = size;
}
// 设置任务队列中任务数量最大值
void ThreadPool::setTaskQueMaxSize(int size)
{
	if (this->checkPoolRunning())
		return;
	this->taskQueMaxSize_ = size;
}
// 向任务队列中提交任务
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
	// 获取互斥锁
	std::unique_lock<std::mutex> lock(taskQueMtx_);
	// 查看队列是否已满, 如果等待1s钟后超时会返回false,输出任务队列已满过一会提交
	
	if (!notFull_.wait_for(lock, 
		std::chrono::seconds(1), 
		[&]() { return curTaskQueSize_ < taskQueMaxSize_; }))
	{
		std::cout << "taskQue is Fulled, submit task is failed" << std::endl;
		return Result(sp, false);
	}
	// 从等待状态到阻塞状态获取倒锁切换成就绪状态准备运行下列代码
	this->taskList_.emplace(sp);
	this->curTaskQueSize_++;
	// 通知所有等待在notempty变量上的线程从wait变成blocking
	notEmpty_.notify_all();
	// 如果是cached模式，任务队列中任务数量大于线程数量，需要动态创建线程
	/*
		1. cached模式
		2. 任务队列中任务数量大于线程池中空闲线程数量
		3. 任务队列中任务数量小于最大任务数量
	*/
	if (this->mode_ == Pool_Mode::CACHED_MODE 
		&& this->curTaskQueSize_ > this->idleThreadPoolSize_ 
		&& this->curTaskQueSize_ < this->taskQueMaxSize_)
	{
		std::cout << "create new thread..." << std::endl;
		// 向线程池容器中创建线程
		std::unique_ptr<Thread> ptr = std::make_unique<Thread>(std::bind(&ThreadPool::ThreadFunc, 
			this, std::placeholders::_1));
		int threadId = ptr->getID();
		threadList_.emplace(threadId, std::move(ptr));
		// 当前线程数量+1
		this->curThreadPoolSize_++;
		// 空闲线程数量+1
		this->idleThreadPoolSize_++;
		// 启动线程
		this->threadList_[threadId]->start();
	}
	return Result(sp, true);
}

// 开启线程池
void ThreadPool::startPool(int size)
{
	// 开启线程池
	this->isPoolRunning_ = true;
	// 设置线程池中线程个数
	this->initThreadPoolSize_ = size;
	// 将Thread类的智能指针存放在容器中
	for (int i = 0; i < this->initThreadPoolSize_; ++i)
	{
		std::unique_ptr<Thread> ptr = std::make_unique<Thread>(std::bind(&ThreadPool::ThreadFunc, this, std::placeholders::_1));
		// 将线程ID与Thread类指针存放在容器中
		this->threadList_.emplace(ptr->getID(), std::move(ptr));
	}
	// 开启每一个线程
	for (int i = 0; i < this->initThreadPoolSize_; ++i)
	{
		this->threadList_[i]->start();
		// 当前线程个数与空闲线程个数分别++
		this->curThreadPoolSize_++;
		this->idleThreadPoolSize_++;
	}
}

// 线程池成员方法
/*
ThreadPool退出作用域需要回收资源，线程对象有以下三种状态
1. 线程对象等待在notEmpty条件变量上
2. 线程对象正在执行任务
3. 线程对象刚执行完任务后，马上有进入了新的循环，但还没抢到锁
在while循环中使用双重判断， while循环判断+if判断 防止第三种情况，任务抢到锁以后等待在notempty变量上
*/
void ThreadPool::ThreadFunc(int threadId)
{
	// 记录当前线程从开始执行到执行完一个任务后的时间戳，用于后续统计线程池中空闲线程时间，并主动释放
	auto lastTime = std::chrono::high_resolution_clock().now();
	for (;;)
	{
		std::shared_ptr<Task> sp;
		{
			// 获取互斥锁
			std::unique_lock<std::mutex> lock(taskQueMtx_);
			std::cout << "tid:" << std::this_thread::get_id() << "尝试获取任务..." << std::endl;
			// 查看任务队列是否为空
			while (curTaskQueSize_ == 0)
			{
				// 双重判断防止出现死锁现象
				// 如果线程池已经结束运行了则退出并回收当前线程
				if (!this->isPoolRunning_)
				{
					// 从map中删除当前线程对象
					this->threadList_.erase(threadId);
					// 退出线程函数就是退出线程
					std::cout << "tid:" << std::this_thread::get_id() << "exit..." << std::endl;
					this->curThreadPoolSize_--;
					isExit_.notify_all();
					return;
				}
				if (this->mode_ == Pool_Mode::FIXED_MODE) // 如果是fixed模式
				{
					notEmpty_.wait(lock);
				}
				else // 如果是cached模式
				{
					// 判断当前线程对象是否等待时间超过了60s
					if (std::cv_status::timeout ==
						notEmpty_.wait_for(lock, std::chrono::seconds(1)))
					{
						// 超时1s钟 去判断当前线程对象的空闲时间
						auto now = std::chrono::high_resolution_clock().now();
						auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
						// 如果超过60s并且当前线程数量高于初始化线程数量，需要将空闲线程进行回收
						if (dur.count() >= THREAD_WAIT_TIMEOUT 
							&& this->curThreadPoolSize_ > this->initThreadPoolSize_)
						{
							// 空闲线程数量-1， 当前线程数量-1
							this->idleThreadPoolSize_--;
							this->curThreadPoolSize_--;
							// 如果超过了60s 则去回收这个线程
							this->threadList_.erase(threadId);
							std::cout << "tid:" << std::this_thread::get_id() << "exit..." << std::endl;
							return;
						}
					}
				}

			}
		} // 释放互斥锁
		// 获取到了任务以后，空闲线程数量-1，当前任务数量-1
		std::cout << "tid:" << std::this_thread::get_id() << "成功获取任务..." << std::endl;
		sp = taskList_.front();
		taskList_.pop();
		// 当前任务数量-1
		curTaskQueSize_--;
		if (nullptr != sp)
		{
			// 空闲线程数量-1
			idleThreadPoolSize_--;
			sp->exec();
		}
		// 如果任务队列中还有其他任务 通知等待在notempty变量上的线程去执行
		if (curTaskQueSize_ > 0)
			this->notEmpty_.notify_all();
		notFull_.notify_all();
		// 更新时间
		lastTime = std::chrono::high_resolution_clock().now();	
	}
	
}

// 查看线程池是否已经开启
bool ThreadPool::checkPoolRunning(void) const
{
	return isPoolRunning_;
}


/* Thread类 */

// 类外初始化静态成员变量
int Thread::generateId_ = 0;

// 构造函数
Thread::Thread(Thread_Func func)
	:threadId_(generateId_++),
	func_(func)
{

}

// 获取线程ID
int Thread::getID() const
{
	return this->threadId_;
}

// 开启线程
void Thread::start()
{
	// 启动一个线程,并执行线程入口函数
	std::thread t(this->func_, this->getID());
	// 设置分离线程
	t.detach();
}

/* Semaphore类 */
Semaphore::Semaphore()
	:resourceLimit_(0)
{

}

// 获取信号量资源
void Semaphore::wait()
{
	// 获取互斥锁
	std::unique_lock<std::mutex> lock(semMtx_);
	// 获取信号量计数
	semCv_.wait(lock, [&]() -> bool { return resourceLimit_ > 0; });
	// 引用计数--
	resourceLimit_--;
}
// 释放信号量资源
void Semaphore::post()
{
	// 获取互斥锁
	std::unique_lock<std::mutex> lock(semMtx_);
	// 引用计数++
	resourceLimit_++;
	// 唤醒等待
	semCv_.notify_all();
}

/* Result类 */
Result::Result(std::shared_ptr<Task> sp, bool flag)
	:task_(sp), isValid_(flag)
{
	// 将当前Result类指针赋给task对象
	task_->setResult(this);
}
// 设置返回值
void Result::setVal(Any any)
{
	// 给any_赋值
	any_ = std::move(any);
	// 资源计数++
	sem_.post();
}
// 获取返回值
Any Result::get()
{
	if (!isValid_)
		return "";
	// 等待信号量资源
	sem_.wait();
	// 返回值
	return std::move(any_);
}

/* Task类 */
void Task::setResult(Result* res)
{
	res_ = res;
}

void Task::exec()
{
	this->res_->setVal(run());
}